package com.chief.transform;


import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class TransformRollingSplit {

    public static void main(String[] args) throws Exception {

        //environment,此处使用连接远程服务器的方式，不使用打包上传
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.createRemoteEnvironment(
                "hadoop52", 8081, "E:\\bigdata-start\\flink\\flinkCode\\target\\flink-1.0-SNAPSHOT.jar");


        //source
        DataStreamSource<String> sourceMap = environment.readTextFile("viewfs://mycluster/data1/rolling.txt");

        //transform
        DataStream<People> map = sourceMap.map(line -> {
            String[] split = line.split(" ");
            return new People(split[0], Integer.valueOf(split[1]), split[2]);
        });

        OutputTag<People> bigger = new OutputTag<>("bigger", TypeInformation.of(People.class));
        OutputTag<People> small = new OutputTag<>("small", TypeInformation.of(People.class));
        OutputTag<People> all = new OutputTag<>("all", TypeInformation.of(People.class));

        SingleOutputStreamOperator<Object> process = map.process(new ProcessFunction<People, Object>() {
            @Override
            public void processElement(People value, Context ctx, Collector<Object> out) throws Exception {
                if (value.getAge() >= 50) {
                    ctx.output(bigger, value);
                } else {
                    ctx.output(small, value);
                }
                ctx.output(all, value);
            }
        });

        //sink
        process.getSideOutput(bigger).print("bigger split");
        process.getSideOutput(small).print("small split");
        process.getSideOutput(all).print("all split");


        environment.execute("transform rolling test");

    }
}
